package cn.lsh.spark;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.io.Serializable;
import java.util.Comparator;

/**
 * 二次排序，有两种方式：
 *  1、先转换成JavaPairRDD，整行作为key，然后通过sortByKey排序，需要自己实现一个Comparator比较器，注意，比较器需要实现Serializable接口
 *  2、自定义一个封装对象，实现Comparable和Serializable接口，然后转换封装对象的JavaRDD，根据sort排序就可以了
 *
 */
public class SecondSort {

	public static void main(String[] args) {
		SparkConf conf = new SparkConf();
		conf.setMaster("local").setAppName("secondSort");

		JavaSparkContext jsc = new JavaSparkContext(conf);

		JavaRDD<String> lines = jsc.textFile("file:/bigdata/hadoop-test/input/spark/secondSort.txt");
		System.out.println("***********第一种方式************");
		JavaPairRDD<String, String> stringStringJavaPairRDD = lines.mapToPair(line -> new Tuple2<>(line, ""));
		stringStringJavaPairRDD.sortByKey(new SecondSortComparator(), false).foreach(o -> System.out.println(o._1()));

		System.out.println("***********第二种方式************");
		lines.map(line -> new SecondSortKey(Integer.parseInt(line.split(" ")[0]), Integer.parseInt(line.split(" ")[1])))
				.sortBy(o -> o, false, 1).foreach(o -> System.out.println(o));

		jsc.stop();
	}

	public static class SecondSortComparator implements Comparator<String>, Serializable {

		@Override
		public int compare(String o1, String o2) {
			String[] ss1 = o1.split(" ");
			String[] ss2 = o2.split(" ");
			int i = Integer.compare(Integer.parseInt(ss1[0]), Integer.parseInt(ss2[0]));
			if (i == 0) {
				return Integer.compare(Integer.parseInt(ss1[1]), Integer.parseInt(ss2[1]));
			}
			return i;
		}
	}

	public static class SecondSortKey implements Comparable<SecondSortKey>, Serializable {
		private int first;
		private int second;

		public SecondSortKey() {

		}

		public SecondSortKey(int first, int second) {
			this.first = first;
			this.second = second;
		}

		public int getFirst() {
			return first;
		}

		public void setFirst(int first) {
			this.first = first;
		}

		public int getSecond() {
			return second;
		}

		public void setSecond(int second) {
			this.second = second;
		}

		@Override
		public String toString() {
			return first + " " + second;
		}

		@Override
		public int compareTo(SecondSortKey o) {
			int i = Integer.compareUnsigned(this.first, o.getFirst());
			if (i == 0) {
				i = Integer.compareUnsigned(this.second, o.getSecond());
			}
			return i;
		}
	}
}

